package org.sean.framework.kafka.logging.init;

import lombok.extern.slf4j.Slf4j;
import org.sean.framework.kafka.logging.LogConstants;
import org.sean.framework.util.IPUtil;
import org.sean.framework.util.ObjectUtil;
import org.sean.framework.util.StringUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.bootstrap.config.BootstrapPropertySource;
import org.springframework.cloud.bootstrap.config.PropertySourceLocator;
import org.springframework.cloud.bootstrap.config.SimpleBootstrapPropertySource;
import org.springframework.context.ApplicationContextInitializer;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.AnnotationAwareOrderComparator;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.EnumerablePropertySource;
import org.springframework.core.env.PropertySource;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Optional;

/**
 * 初始化Kafka配置
 *
 * @author xielei
 */
@Slf4j
public class KafkaLogInitializer implements ApplicationContextInitializer<ConfigurableApplicationContext>, Ordered {

    @Autowired(required = false)
    private List<PropertySourceLocator> propertySourceLocators = new ArrayList<>();

    @Override
    public void initialize(ConfigurableApplicationContext applicationContext) {
        List<PropertySource<?>> composite = new ArrayList<>();
        AnnotationAwareOrderComparator.sort(this.propertySourceLocators);
        ConfigurableEnvironment environment = applicationContext.getEnvironment();
        for (PropertySourceLocator locator : this.propertySourceLocators) {
            Collection<PropertySource<?>> source = locator.locateCollection(environment);
            if (ObjectUtil.isEmpty(source)) {
                continue;
            }
            List<PropertySource<?>> sourceList = new ArrayList<>();
            for (PropertySource<?> p : source) {
                if (p instanceof EnumerablePropertySource) {
                    EnumerablePropertySource<?> enumerable = (EnumerablePropertySource<?>) p;
                    sourceList.add(new BootstrapPropertySource<>(enumerable));
                } else {
                    sourceList.add(new SimpleBootstrapPropertySource<>(p));
                }
            }
            composite.addAll(sourceList);
        }

        composite.forEach(item -> environment.getPropertySources().addFirst(item));

        String name = environment.getProperty("spring.application.name");
        String port = environment.getProperty("server.port");
        if (StringUtil.isEmpty(port)) {
            port = Optional.ofNullable(IPUtil.getServerPort()).orElse(0).toString();
        }
        String servers = environment.getProperty(LogConstants.KAFKA_SERVERS);
        String topic = environment.getProperty(LogConstants.KAFKA_TOPIC);
        if (!System.getProperties().containsKey(LogConstants.KAFKA_SERVERS_PROPERTY)
                && StringUtil.isNotEmpty(servers) && StringUtil.isNotEmpty(topic)) {
            System.setProperty(LogConstants.KAFKA_SERVERS_PROPERTY, servers);
            System.setProperty(LogConstants.KAFKA_TOPIC_PROPERTY, topic);
            System.setProperty(LogConstants.KAFKA_CLIENT_ID_PROPERTY, name + IPUtil.getIP() + port);
        }
    }

    @Override
    public int getOrder() {
        return Ordered.HIGHEST_PRECEDENCE;
    }
}
